/*
 *  Copyright 2008 The MITRE Corporation (http://www.mitre.org/). All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */package org.mitre.mrald.query;

import com.sun.rowset.WebRowSetImpl;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.Options;
import org.apache.axis2.rpc.client.RPCServiceClient;
import org.mitre.mrald.control.MsgObject;
import org.mitre.mrald.output.OutputManager;
import org.mitre.mrald.parser.ParserElement;
import org.mitre.mrald.util.Config;
import org.mitre.mrald.util.MraldOutFile;
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.io.IOException;
import java.io.StringReader;
import java.net.URL;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.sql.rowset.WebRowSet;
import javax.xml.namespace.QName;


/**
 * Central class for managing the broadcast query capability.  Manages the BroadcastQueryProvider as well as doing
 * the actual broadcasting and collecting.
 *
 * @author $author$
 * @version $Revision: 1.4 $
 */
public class BroadcastQuery
{
    private static PropertyChangeListener pcl;

    /** Provider property. */
    private static BroadcastTargetProvider provider = null;

    /* Here be stuff for doing the broadcasts */
    /** The Map of currently stored BroadcastManagers */
    public static final Vector userThreads = new Vector(  );

    /** DOCUMENT ME! */
    protected static Hashtable<MsgObject, BroadcastManager> activeBroadcasts = new Hashtable<MsgObject, BroadcastManager>(  );

    /**
     * Description of the Method
     */
    public static void init(  )
    {
        if ( pcl == null )
        {
            pcl = ( new PropertyChangeListener(  )
                    {
                        public void propertyChange( PropertyChangeEvent evt )
                        {
                            loadTargets(  );
                        }
                    } );
            Config.addPropertyChangeListener( BroadcastQuery.pcl );
        }
    }

    /**
     * DOCUMENT ME!
     */
    protected static void loadTargets(  )
    {
        provider.reload(  );
    }

    /**
     * Get provider property.
     *
     * @return Provider property.
     */
    public static BroadcastTargetProvider getProvider(  )
    {
        try
        {
            Object p = provider.getClass(  ).newInstance(  );

            return ( BroadcastTargetProvider ) p;
        }
        catch ( Exception e )
        {
            MraldOutFile.logToFile( e );
        }

        return null;
    }

    /**
     * Set provider property.
     *
     * @param _provider New provider property.
     */
    public static void setProvider( BroadcastTargetProvider _provider )
    {
        provider = _provider;
        loadTargets(  );
    }

    /**
     * DOCUMENT ME!
     *
     * @param msg DOCUMENT ME!
     * @param targetList DOCUMENT ME!
     */
    public static void doBroadcast( MsgObject msg, List<BroadcastQueryTarget> targetList )
    {
        // MraldOutFile.appendToFile( "In doBroadcast() with # targets = " + targetList.size() );
        try
        {
            BroadcastManager manager = new BroadcastManager(  );

            //set up broadcast mamanger and put into activeBroadcasts
            for ( BroadcastQueryTarget target : targetList )
            {
                try
                {
                    manager.add( target );
                }
                catch ( Throwable t )
                {
                    MraldOutFile.logToFile( t );

                    //don't let anything stop the normal processing of the local query
                    continue;
                }
            }

            activeBroadcasts.put( msg, manager );

            //BroadcastManager should be the one to kick off all the Broadcasts
        }
        catch ( Throwable t )
        {
            MraldOutFile.logToFile( t );

            //don't let anything stop the normal processing of the local query
            return;
        }
    }

    /**
     * DOCUMENT ME!
     *
     * @param msg DOCUMENT ME!
     * @param output DOCUMENT ME!
     */
    public static void processBroadcast( MsgObject msg, OutputManager output )
    {
        BroadcastManager manager = activeBroadcasts.get( msg );
        manager.processWithOuputManager( msg, output );
    }
}


/**
 * This class is responsible for managing all the broadcasts to other MRALD instances.  It provides methods to
 * build up a group of broadcasts, get the result set
 *
 * @author $author$
 * @version $Revision: 1.4 $
 */
class BroadcastManager
{
    /** DOCUMENT ME! */
    protected List<Future<ResultSet>> workingBroadcasts = new ArrayList<Future<ResultSet>>(  );

    /** DOCUMENT ME! */
    protected List<Future<ResultSet>> doneBroadcasts = null;

    /** DOCUMENT ME! */
    protected Map<Future<ResultSet>, BroadcastQueryTarget> sources = new HashMap<Future<ResultSet>, BroadcastQueryTarget>(  );

    /** DOCUMENT ME! */
    protected Map<Future<ResultSet>, String> queries = new HashMap<Future<ResultSet>, String>(  );

    /** DOCUMENT ME! */
    protected ExecutorService executer = Executors.newCachedThreadPool(  );

    /**
     * DOCUMENT ME!
     *
     * @param target DOCUMENT ME!
     */
    @SuppressWarnings("unchecked")
	public void add( BroadcastQueryTarget target )
    {
        String query = "";
        MsgObject msg = new MsgObject(  );
        msg.setWorkingObjects( new ArrayList<ParserElement>( Arrays.asList( target.getSqlElements(  ) ) ) );
        msg.setLinks( new ArrayList<LinkElement>( Arrays.asList( target.getLinkElements(  ) ) ) );

        QueryBuilder builder = new QueryBuilder( msg );

        try
        {
            builder.buildQueryComponents( msg.getWorkingObjects(  ) );
            query = builder.buildQuery(  );
        }
        catch ( Throwable t )
        {
            MraldOutFile.logToFile( t );

            //don't let anything stop the normal processing of the local query
            return;
        }

        Broadcast doer = new Broadcast( target.getUrl(  ), target.getRemoteDatasource(  ), query );

        /*
         *  The return value of submit()--and this is the part I love--is a Future object: essentially an IOU for a result
         *  some time in the future. When I'm ready to use the result of my task, I simply call the get() method of the
         *  Future object. If the task is done executing, then get() returns my result right away. Otherwise, it blocks
         *  until the result is ready. If Callable throws an exception, then the get() method wraps that exception in an
         *  ExecutionException and throws that. Future also has methods for cancelling and querying the status of task
         *  execution, but you'll have to look those up yourself.
         */
        // PM: Assume that doer is of the appropriate type.
        Future<ResultSet> p = executer.submit( doer );
        workingBroadcasts.add( p );
        sources.put( p, target );
        queries.put( p, query );
    }

    /**
     * Moves completed queries from the working to done ArrayLists
     */
    protected void checkDone(  )
    {
        doneBroadcasts = new ArrayList<Future<ResultSet>>(  );

        synchronized ( workingBroadcasts )
        {
            for ( Future<ResultSet> future : workingBroadcasts )
            {
                if ( future.isDone(  ) )
                {
                    doneBroadcasts.add( future );
                }
            }

            workingBroadcasts.removeAll( doneBroadcasts );
        }
    }

    /**
     * DOCUMENT ME!
     *
     * @param msg DOCUMENT ME!
     * @param output DOCUMENT ME!
     */
    public void processWithOuputManager( MsgObject msg, OutputManager output )
    {
        int waitTime = 0;
        int maxWaitTime = 0;
        String max = Config.getProperty( "broadcastWaitSeconds" );

        try
        {
            maxWaitTime = Integer.parseInt( max );
        }
        catch ( Exception e )
        {
            maxWaitTime = 60;
        }

        checkDone(  );

        while ( ( ( workingBroadcasts.size(  ) + doneBroadcasts.size(  ) ) > 0 ) && ( waitTime < maxWaitTime ) )
        {
            for ( Future<ResultSet> future : doneBroadcasts )
            {
                try
                {
                    ResultSet rs = future.get(  );

                    if ( rs != null )
                    {
                        output.setResultSet( rs );
                        output.setQuery( queries.get( future ) );
                        output.printNewResultHeader( sources.get( future ) );
                        output.printBody(  );
                        output.printLimit(  );
                    }
                }
                catch ( Throwable e )
                {
                    //do NOT hold up this output because of a remote problem
                    MraldOutFile.logToFile( e );
                }
            }

            if ( workingBroadcasts.size(  ) > 0 )
            {
                try
                {
                    TimeUnit.SECONDS.sleep( 2 );
                }
                catch ( InterruptedException e )
                {
                }

                waitTime += 2;
            }

            checkDone(  );
        }
    }
}


/**
 * This class will connect to another MRALD instance and retrieve query results
 *
 * @author $author$
 * @version $Revision: 1.4 $
 */
class Broadcast implements Callable
{
    /** Url property. */
    protected URL url = null;

    /** Query property. */
    protected String query = null;

    /** Datasource property. */
    protected String datasource = null;

/**
     * Creates a new Broadcast object.
     *
     * @param url URL of MRALD instance
     * @param query The query to be answered
     */
    public Broadcast( URL url, String datasource, String query )
    {
        this.url = url;
        this.query = query;
        this.datasource = datasource;
    }

    /**
     * Get datasource property.
     *
     * @return Datasource property.
     */
    public String getDatasource(  )
    {
        return this.datasource;
    }

    /**
     * Set datasource property.
     *
     * @param datasource New datasource property.
     */
    public void setDatasource( String datasource )
    {
        this.datasource = datasource;
    }

    /**
     * Get url property.
     *
     * @return Url property.
     */
    public URL getUrl(  )
    {
        return this.url;
    }

    /**
     * Set url property.
     *
     * @param url New url property.
     */
    public void setUrl( URL url )
    {
        this.url = url;
    }

    /**
     * Get query property.
     *
     * @return Query property.
     */
    public String getQuery(  )
    {
        return this.query;
    }

    /**
     * Set query property.
     *
     * @param query New query property.
     */
    public void setQuery( String query )
    {
        this.query = query;
    }

    /**
     * Main processing method for the Broadcast object.  Broadcasts query to one remote MRALD instance and
     * returns the ResultSet.
     *
     * @return ResultSet of query
     *
     * @throws IOException DOCUMENT ME!
     * @throws SQLException DOCUMENT ME!
     */
    public ResultSet call(  ) throws IOException, SQLException
    {
        /*
         *  set up the action to take in the web service
         */
        QName actionQname = new QName( "http://webservices.mrald.mitre.org/xsd", "handleBroadcastQuery" );

        RPCServiceClient serviceClient = new RPCServiceClient(  );
        Options options = serviceClient.getOptions(  );
        EndpointReference targetEPR = new EndpointReference( url + "/services/HandleBroadcastQuery" );
        options.setTo( targetEPR );
        options.setAction( actionQname.getLocalPart(  ) );
        serviceClient.setOptions( options );

        /*
         *  setting up the stuff to pass to the web service
         */
        Object[] args = new Object[] { datasource, query };

        /*
         *  set return type to be an array of Query objects
         */
        Class[] returnTypes = new Class[] { String.class };

        /*
         *  do the web service call
         */
        Object[] response = serviceClient.invokeBlocking( actionQname, args, returnTypes );

        if ( ( response[0] == null ) || ( response.length == 0 ) )
        {
            MraldOutFile.appendToFile( "No answer received!" );

            return null;
        }

        String xmlString = ( String ) response[0];
        StringReader reader = new StringReader( xmlString );
        WebRowSet wrs = new WebRowSetImpl(  );
        wrs.readXml( reader );

        return wrs;
    }
}

